package com.lagou.rpc.consumer.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.common.MyZkSerializer;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import lombok.Data;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

import static com.lagou.rpc.constant.Constant.PROVIDER_PATH;
import static com.lagou.rpc.constant.Constant.ZOOKEEPER_ADDRESS;

/**
 * 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 */
@Component
@Data
public class ZookeeperWatcher {

    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;

    private static List<String> servers = new ArrayList<>();

    private static ConcurrentHashMap<String, String> serverMap = new ConcurrentHashMap<>();

    private int requestCount = 0;

    private static ZkClient zkClient;

    static {
        zkClient = new ZkClient(ZOOKEEPER_ADDRESS);
        zkClient.setZkSerializer(new MyZkSerializer());
        System.out.println("会话被创建了..");

        // 获取子节点列表
        List<String> children = zkClient.getChildren(PROVIDER_PATH);
        System.out.println(children);
        setServices(children);

        zkClient.subscribeChildChanges(PROVIDER_PATH, (parentPath, list) -> {
            System.out.println(parentPath + "的子节点列表发生了变化,变化后的子节点列表为" + list);
            reSetServices(list);
        });
    }

    public Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, (proxy, method, args) -> {
                    //1.封装request请求对象
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setRequestId(UUID.randomUUID().toString());
                    rpcRequest.setClassName(method.getDeclaringClass().getName());
                    rpcRequest.setMethodName(method.getName());
                    rpcRequest.setParameterTypes(method.getParameterTypes());
                    rpcRequest.setParameters(args);
                    //2.创建RpcClient对象
                    RpcClient rpcClient = null;
                    String providerPath = "";
                    for (int i = 0; i < servers.size(); i++) {
                        //获取负载服务索引
                        int serverIndex = (requestCount++) % servers.size();
                        providerPath = servers.get(serverIndex);
                        String[] server = providerPath.split(":");
                        try {
                            rpcClient = new RpcClient(server[0], Integer.parseInt(server[1]));
                            System.out.println(providerPath + "RPC客户端已创建!");
                            break;
                        } catch (Exception exception) {
                            System.out.println("连接异常....尝试其他服务地址");
                            //连接异常移除地址
                            servers.remove(serverIndex);
                            i--;//重新轮询
                        }
                    }
                    try {
                        //3.发送消息
                        long start = System.currentTimeMillis();
                        assert rpcClient != null;
                        Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                        long end = System.currentTimeMillis();
                        long responseTime = end - start;
                        // 设置响应时间
                        zkClient.writeData(serverMap.get(providerPath), System.currentTimeMillis() + "," + responseTime + "," + providerPath);
                        RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                        if (rpcResponse.getError() != null) {
                            throw new RuntimeException(rpcResponse.getError());
                        }
                        //4.返回结果
                        Object result = rpcResponse.getResult();
                        return JSON.parseObject(result.toString(), method.getReturnType());
                    } finally {
                        assert rpcClient != null;
                        rpcClient.close();
                    }

                });
    }

    @PostConstruct
    public void startTask() {
        //开始定时任务  5秒执行一次
        threadPoolTaskScheduler.scheduleAtFixedRate(() -> {
            System.out.println("查找最小耗时provider");
            long currentTimeMillis = System.currentTimeMillis();
            int minResponseTime = 5000;
            String minProviderPath = "";
            for (String server : servers) {
                String zookeeperNode = serverMap.get(server);
                String data = zkClient.readData(zookeeperNode);
                String[] split = data.split(",");
                long timeStamp = Long.parseLong(split[0]);
                int responseTime = Integer.parseInt(split[1]);
                String providerPath = split[2];
                // 超过5秒没调用就重置
                if ((currentTimeMillis - timeStamp) > 5000) {
                    reSetServices(zkClient.getChildren(PROVIDER_PATH));
                    return;
                }
                if (responseTime < minResponseTime){
                    minResponseTime = responseTime;
                    minProviderPath = providerPath;
                }
            }
            servers.clear();
            servers.add(minProviderPath);
            System.out.println("最小耗时provider为 : " + servers);
        }, 5000);
    }

    /**
     * 没有响应时间时重置serviceList
     */
    private static void reSetServices(List<String> children) {
        System.out.println("重置services");
        servers.clear();
        serverMap.clear();
        setServices(children);
    }

    /**
     * 设置serviceList
     */
    private static void setServices(List<String> children) {
        for (String child : children) {
            String path = PROVIDER_PATH + "/" + child;
            boolean exists = zkClient.exists(path);
            if (exists) {
                String o = zkClient.readData(path);
                String[] split = o.split(",");
                servers.add(split[2]);
                serverMap.put(split[2], path);
            }
        }
        System.out.println("设置services为 : " + servers);
    }

}
